package dk.almbrand.microservice.kafka.producer;

import dk.almbrand.microservice.exception.SystemException;

import java.util.Properties;

import kafka.javaapi.producer.Producer;

import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class ProducerImpl implements dk.almbrand.microservice.kafka.producer.Producer {

    final static String TOPIC = "verification-topic";

    private String topic;
    private String host;
    private String serializer;
    private Producer<String, String> producer = null;

    public ProducerImpl() {       
    }

    public void publish(String topic, String key, String data) throws SystemException {
        try {
            new Producer<String, String>(getKafkaConfiguration()).send(new KeyedMessage(TOPIC, key, data));
        } catch (Exception e) {
            System.out.println("Could not send data: " + data + " topic: " + topic + " error msg: " + e.getMessage());
            throw new SystemException("Could not send data: " + data + " topic: " + topic, e);
        } finally {
            if(producer != null) {
                producer.close();
            }
        }
    }
    
    private ProducerConfig getKafkaConfiguration() {
        Properties props = new Properties();
        props.put("metadata.broker.list", host); // "ubuntu:9092";
        props.put("serializer.class", serializer); //"kafka.serializer.StringEncoder");
        props.put("zookeeper.session.timeout.ms", "5000");
        props.put("request.timeout.ms", "5000");
        props.put("request.required.acks", "1");
        
        //   props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
           
        return new ProducerConfig(props);
    }
    
    public ProducerImpl build() {
        producer = new Producer<String, String>(getKafkaConfiguration());
        return this;
    }

    public ProducerImpl setHost(String host) {
        this.host = host;
        return this;
    }

    public String getHost() {
        return host;
    }

    public ProducerImpl setSerializer(String serializer) {
        this.serializer = serializer;
        return this;
    }

    public String getSerializer() {
        return serializer;
    }
    
    public ProducerImpl setTopic(String topic) {
        this.topic = topic;
        return this;
    }

    public String getTopic() {
        return topic;
    }
    
    public static void main(String[] args) throws SystemException {
        for (int i = 0; i <= 100; i++) {
            new ProducerImpl().setHost("ubuntu:9092").setSerializer("kafka.serializer.StringEncoder").publish(TOPIC, "", "data test: " + i);
        }
    }
}
